package com.example.observer;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.NavigableMap;

/**
 * @author lbq
 * @date 2021/11/28 10:31
 */
public class MyObserver extends BaseRegionObserver {

    private final byte[] family = Bytes.toBytes("friends");

    @Override
    public void postDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        final HTableInterface relation = e.getEnvironment().getTable(TableName.valueOf("relation"));

        // 获取要删除的好友关系
        final NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();
        for (List<Cell> cells : familyCellMap.values()) {
            if (cells == null || cells.size() == 0) {
                continue;
            }
            for (Cell cell : cells) {
                //rowkey信息
                byte[] rowkey = CellUtil.cloneRow(cell);
                //列信息
                byte[] column = CellUtil.cloneQualifier(cell);

                //验证删除的目标数据是否存在，存在则执行删除否则不执行,必须有此判断否则造成协处理器被循环调用耗尽资源
                boolean flag = relation.exists(new Get(column).addColumn(family, rowkey));
                if (flag) {
                    Delete newDelete = new Delete(column).addColumn(family, rowkey);
                    relation.delete(newDelete);
                }
            }
        }
        relation.close();
    }

    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
        final HTableInterface relation = e.getEnvironment().getTable(TableName.valueOf("relation"));

        final NavigableMap<byte[], List<Cell>> familyCellMap = put.getFamilyCellMap();

        for (List<Cell> cells : familyCellMap.values()) {
            if (cells == null || cells.size() == 0) {
                continue;
            }
            for (Cell cell : cells) {
                //rowkey信息
                byte[] rowkey = CellUtil.cloneRow(cell);
                //列信息
                byte[] column = CellUtil.cloneQualifier(cell);

                //验证插入的目标数据是否存在，不存在则执行插入,否则不执行
                boolean flag = relation.exists(new Get(column).addColumn(family, rowkey));
                if (!flag) {
                    Put newPut = new Put(column).addColumn(family, rowkey, rowkey);
                    relation.put(newPut);
                }
            }
        }
        relation.close();
    }
}
